package trunk.firstSec;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

import java.io.*;
import java.util.ArrayList;

public class daMapperNgoNgo extends MapReduceBase implements Mapper<IntWritable, BytesWritable, Text, IntWritable>{
    private static ArrayList<byte[]> query = new ArrayList<>(0);

    public void configure(JobConf job){
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader( new FileInputStream(new File(DistributedCache.getCacheFiles(job)[0].getPath()))));
            String line;
            while((line = reader.readLine())!=null)
            {
                query.add(line.getBytes());
            }


        } catch (IOException e){ e.printStackTrace(); }
    }
    @Override
    public void map(IntWritable keyIn, BytesWritable valueIn, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        byte[] input = valueIn.getBytes();  int blockStart = keyIn.get();
        Text keyOut = new Text();           IntWritable valueOut = new IntWritable();   int patternLength;
        String currentFile = ((FileSplit)reporter.getInputSplit()).getPath().getName();

        for (byte[] currentQuery : query) {
            keyOut.set(currentFile + "," + new String(currentQuery));
            output.collect(keyOut, new IntWritable(-1)); // try make every query appear

            int root = 0,   runner = 0, length = currentQuery.length - 1;
            while (root + runner != input.length-1){
                if (currentQuery[runner] == input[(root + runner)])       // if char at 'i' = char in query [base loc + i]
                {
                    if (runner == length)  // if exact match
                    {
                        valueOut.set(blockStart + root);    // adjust + 1 here if needed
                        output.collect(keyOut, valueOut);
                        root++;
                        runner = 0;
                    }
                    else                        // if kinda gonna match maybe?
                        runner++;
                }
                else
                {
                    root++;
                    runner = 0;
                }
            }
        }
    }
}
